package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo22Acc {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo22Acc")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val intRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9))

    var cnt: Int = 0

    /**
     * 实际上在一个Spark程序当中，虽然代码都在一个main方法中
     * 但实际运行时会分为两个部分：
     * 算子外部的代码在Driver端运行
     * 算子内部的代码会被封装成Task然后由Driver端发送给Executor去执行
     *
     * 如果在算子内部使用算子外部的变量时，实际上用到的是该变量的副本
     * 如果直接对该外部变量进行修改，实际上修改的是该变量的副本，Driver的变量不会被修改
     *
     * 如果需要在算子内部对外部的变量进行修改并且生效，则需要使用累加器
     *
     * 累加器的使用：
     * 在算子外部定义，在算子内部使用，在Driver进行汇总
     */

    intRDD.map(i => {
      cnt += 1
      println(s"在map方法中的cnt的值为：$cnt")
      i
    }).count()

    cnt += 1

    println(s"在map方法结束之后的cnt的值为：$cnt")

    // 在算子外部即Driver端定义一个Long类型的累加器变量，默认初始值为0
    val longAcc: LongAccumulator = sc.longAccumulator("LongAcc")

    val intRDD2: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11), 3)
    intRDD2.map(i => {
      // 在算子内部进行累加
      longAcc.add(1)
      println(s"当前在map算子内部的累加器值为：${longAcc.value}")
      i
    }).count()

    // 在Driver端汇总
    println(s"在Driver端汇总之后的累加器值为：${longAcc.value}")


  }

}
